package com.wuqing.consumer.utils;

import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import org.apache.dubbo.common.utils.NamedThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.*;

/**
 * @author wuqing
 * @date 2020/8/8
 */
public class AsyncExecutor {

    private static final Logger logger = LoggerFactory.getLogger(AsyncExecutor.class);

    private static final int CORE_SIZE = 100;

    private static final BlockingQueue<Runnable> TASK_QUEUE = new LinkedBlockingDeque<>(20000);

    private static final String COMMON_EXECUTOR_PREFIX = "dubbo-executor";

    private static final RejectedExecutionHandler REJECTED_EXECUTION_HANDLER = (r, executor) -> {
        System.out.println("队列已满，抛弃任务......");
        //throw new RejectedExecutionException("Task " + r.toString() + " rejected from " + executor.toString());
    };

    private static final ThreadPoolExecutor EXECUTOR = new ThreadPoolExecutor(CORE_SIZE, CORE_SIZE * 2,
            0, TimeUnit.SECONDS, TASK_QUEUE, new NamedThreadFactory(COMMON_EXECUTOR_PREFIX),
            REJECTED_EXECUTION_HANDLER);

    private static final ListeningExecutorService ASYNC_EXECUTOR = MoreExecutors.listeningDecorator(EXECUTOR);

    static {
        Thread shutdownThread = new Thread("destroyExecutorThread") {
            @Override
            public void run() {
                try {
                    ASYNC_EXECUTOR.shutdown();
                    ASYNC_EXECUTOR.awaitTermination(5, TimeUnit.SECONDS);
                } catch (InterruptedException e) {
                    logger.error("5s内线程池仍未关闭掉.........");
                    Thread.currentThread().interrupt();
                }
            }
        };
        Runtime.getRuntime().addShutdownHook(shutdownThread);
    }

    public static ListeningExecutorService getExecutor() {
        return ASYNC_EXECUTOR;
    }

    public static <T> ListenableFuture<T> call(Callable<T> callable) {
        Preconditions.checkNotNull(callable);
        return ASYNC_EXECUTOR.submit(callable);
    }

    public static void run(Runnable runnable) {
        Preconditions.checkNotNull(runnable);
        ASYNC_EXECUTOR.execute(runnable);
    }

}